package com.zxq.common.middleware.lock;


import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
import com.zxq.common.utils.exception.MultiLockFailException;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

@Slf4j
public class RedisMultiLock implements Lock {

    @Override
    public boolean tryLock() {
        try {
            return tryLock(NO_WAIT_TIME, null);
        } catch (InterruptedException e) {
            log.debug("MultiLock tryLock interrupted. MultiLockName: {}", getName());
        }

        return false;
    }

    @Override
    public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
        if (waitTime < 0 || unit == null) {
            log.warn("parameter not valid, set waitTime to -1, unit = null. {}, {}", waitTime, unit);
            waitTime = NO_WAIT_TIME;
            unit = null;
        }

        long endTime = NO_WAIT_TIME;
        if (unit != null) {
            endTime = System.nanoTime() + unit.toNanos(waitTime);
        }

        boolean lockAcquired;
        List<Lock> acquiredLocks = new ArrayList<>(internalLocks.size());
        //多次尝试获取锁
        while (true) {
            //逐个锁按顺序获取
            for (Lock lock : internalLocks) {
                lockAcquired = false;
                try {
                    if (waitTime == NO_WAIT_TIME) {
                        lockAcquired = lock.tryLock();
                    } else {
                        lockAcquired = lock.tryLock(unit.toMillis(waitTime), TimeUnit.MILLISECONDS);
                    }
                } catch (InterruptedException e) {
                    log.debug("MultiLock tryLock interrupted. MultiLockName: {}", getName());

                    //rollback then return, 响应中断
                    unlockLocksAsync(acquiredLocks, lock);

                    throw new InterruptedException();
                } catch (Exception e) {
                    lockAcquired = false;
                    log.warn("lock.tryLock method error, {}", getName(), e);
                }

                //if trylock failed, must revert locked, then return or retry next round
                if (!lockAcquired) {
                    log.debug("failed to obtain the lock, start rollback. {}", acquiredLocks);

                    //rollback
                    unlockLocksAsync(acquiredLocks, lock);

                    //skip next round of retry
                    if (waitTime == NO_WAIT_TIME) {
                        return false;
                    }

                    if (System.nanoTime() > endTime) {
                        log.debug("MultiLock failed to obtains all the locks within {}ms.", unit.toMillis(waitTime));
                        return false;
                    }

                    //start another round after rollback
                    acquiredLocks.clear();
                    break;
                }

                //
                if (lockAcquired) {
                    // 在成功获取到一个锁后，将该锁添加到 acquiredLocks 列表的第一个位置。这是为了在回滚操作时，能够按照相反的顺序释放锁，从而避免死锁的风险。
                    acquiredLocks.add(0, lock);
                }

                if (acquiredLocks.size() == internalLocks.size()) {
                    log.debug("MultiLock obtained all the locks. {}", getName());

                    return true;
                }

                if (Thread.interrupted()) {
                    //rollback then return, 响应中断
                    unlockLocksAsync(acquiredLocks, lock);

                    throw new InterruptedException();
                }

                if (waitTime != NO_WAIT_TIME && System.nanoTime() > endTime) {
                    log.debug("MultiLock failed to obtains all the locks within {}ms.", unit.toMillis(waitTime));

                    //rollback
                    unlockLocksAsync(acquiredLocks, lock);

                    return false;
                }
            }

            LockSupport.parkNanos(this, TimeUnit.MILLISECONDS.toNanos(DistributeLockConstants.RedisLockConstants.SPIN_AWAIT_TIME));
        }
    }

    @Override
    public void unlock() {
        log.debug("RedisMultiLock.unlock(), lockNames: {}", lockNames);

        unlockLocksAsync(internalLocks, null);
    }

    @Override
    public String getName() {
        StringBuilder sb = new StringBuilder();
        sb.append("RedisMultiLock: ");
        for (String item : lockNames) {
            sb.append(item);
            sb.append(", ");
        }

        if (!lockNames.isEmpty()) {
            sb.delete(sb.length() - 2, sb.length());
        }

        return sb.toString();
    }

    ///////////////////////////////////////////////////////////////////////

    private static final int ESTIMATED_TRY_MILLI_SECOND = 1500;
    private static final int NO_WAIT_TIME = -1;

    private ListeningExecutorService executorService = MoreExecutors.listeningDecorator(ThreadPoolUtils.getMultiLockExecutor());

    final protected List<Lock> internalLocks;
    private List<String> lockNames;

    public RedisMultiLock(List<String> lockNames, RedisProcessor redisProcessor, final RenewalWatchdog watchdog, int expireTime, int retry) {
        if (CollectionUtils.isEmpty(lockNames)) {
            throw new MultiLockFailException("lock list is empty");
        }

        if (lockNames.size() > DistributeLockConstants.MultiLockConstants.MAX_SIZE) {
            throw new MultiLockFailException( "lock list is too long");
        }

        List<Lock> locks = lockNames.stream()
                .map(lockName -> new RedisLock(lockName, redisProcessor, watchdog, expireTime, retry))
                .map(lock -> (Lock)lock)
                .toList();
        this.internalLocks = new ArrayList<>(locks.size());
        this.lockNames = new ArrayList<>(locks.size());
        init(locks);
    }

    public RedisMultiLock(List<Lock> locks) {
        if (CollectionUtils.isEmpty(locks)) {
            throw new MultiLockFailException("lock list is empty");
        }

        if (locks.size() > DistributeLockConstants.MultiLockConstants.MAX_SIZE) {
            throw new MultiLockFailException( "lock list is too long");
        }

        internalLocks = new ArrayList<>(locks.size());
        lockNames = new ArrayList<>(locks.size());
        init(locks);
    }

    private void init(List<Lock> locks) {
        Set<String> lockNameSet = new HashSet<>();
        TreeSet<Lock> lockSet = new TreeSet<Lock>(new Comparator<Lock>() {
            @Override
            public int compare(Lock lock1, Lock lock2) {
                return lock1.getName().compareTo(lock2.getName());
            }
        });

        for (Lock lock : locks) {
            if (!validateLock(lock)) {
                continue;
            }

            if (lockNameSet.contains(lock.getName())) {
                log.error("Duplicate lock error: {}, current locks: {}", lock.getName(), lockNameSet);
                throw new MultiLockFailException("duplicate lock");
            }

            lockSet.add(lock);
            lockNameSet.add(lock.getName());

            log.debug("lock added. {}", lock.getName());
        }

        if (CollectionUtils.isEmpty(lockSet)) {
            throw new MultiLockFailException("lockSet is empty");
        }

        for (Lock tmp : lockSet) {
            internalLocks.add(tmp);
            lockNames.add(tmp.getName());
        }

        log.debug("MultiLock added {} locks", internalLocks.size());
    }

    private boolean validateLock(Lock lock) {
        if (lock == null) {
            return false;
        }

        if (StringUtils.isEmpty(lock.getName())) {
            return false;
        }

        return lock instanceof RedisLock;
    }

    /**
     * returns true while all items inside list equal to true
     *
     * @param list
     * @return
     */
    private boolean checkBooleanList(List<Boolean> list) {
        if (CollectionUtils.isEmpty(list)) {
            return false;
        }

        for (boolean item : list) {
            if (!item) {
                return false;
            }
        }

        return true;
    }

    /**
     * 使用 RedisLock 异步解锁，不关心结果（RedisLock 自带重试逻辑）
     * @param list
     * @param current
     */
    private void unlockLocksAsync(List<Lock> list, Lock current) {
        if (list == null) {
            list = new ArrayList<>();
        }

        if (current != null) {
            list.add(0, current);
        }

        for (Lock lock : list) {
            try {
                if (lock instanceof RedisLock) {
                    log.debug("Starting unlock asynchronously {}", lock.getName());
                    ((RedisLock) lock).unlockAsync();
                }
            } catch (Exception e) {
                log.warn("unlockLocksAsync failed. lockName: {}", lock.getName(), e);
            }
        }
    }

}
